/*
 * Copyright (c) 2016. 恒昌互联网运营中心.
 */

package com.credithc.eagleeye.queue;
import com.alibaba.fastjson.JSONObject;
import com.credithc.eagleeye.agent.util.HostUtil;
import com.credithc.eagleeye.config.EagleMonitorESConfig;
import com.credithc.eagleeye.thread.ThreadFactoryImpl;
import com.credithc.eagleeye.util.FastJsonConvert;
import com.credithc.eagleeye.vo.RpcTraceInfoVo;
import com.credithc.framework.es.util.ElasticSearchUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
/**
 * 描 述: <br/>
 * 作 者: zhangjin<br/>
 * 创 建：2016年06月07日<br/>
 * 版 本：v1.0.0<br>
 * <p/>
 * 历 史: (版本) 作者 时间 注释 <br/>
 */
@Component
@Slf4j
public class EagleTraceInfoQueueManager  {
    @Autowired
    private EagleMonitorESConfig eagleMonitorESConfig;
    @Autowired
    private ElasticSearchUtil esUtil;
    private static AtomicBoolean initialized = new AtomicBoolean(false);
    /**
     * 在eagle-springboot中收集http与redis里的鹰眼日志信息
     * 为防止eagle-eye-boot内存溢出，应为有界队列
     * 假设每条数据1kb，则4GB内存可以容纳1024*1024*4=4194304条
     */
    private volatile LinkedBlockingQueue<RpcTraceInfoVo> queue = new LinkedBlockingQueue<RpcTraceInfoVo>(102400);
    private static ExecutorService singleThreadExecutor =new ThreadPoolExecutor(1,
            1,0L, TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<Runnable>(),
            new ThreadFactoryImpl("sendRpcInfoThread_")
    );

    private AtomicInteger total = new AtomicInteger(0);
    private AtomicInteger fail = new AtomicInteger(0);

    private EagleTraceInfoQueueManager() {
    }

    public synchronized void start()  {
        if( initialized.compareAndSet(false, true)) {
            if (!eagleMonitorESConfig.isEnable()) {
                log.warn("eagle-eye-transfer sendRpcInfo thread is disable!");
                return;
            }
            singleThreadExecutor.submit(new EagleESMonitorQueueTask());
            log.info("eagle-eye-transfer sendRpcInfo thread is start!");
        }
    }

    public synchronized void stop()  {
        singleThreadExecutor.shutdownNow();
        log.info("eagle-eye-transfer sendRpcInfo thread is stop!");
    }

    /**
     * 插入鹰眼数据到队列队尾
     *
     * @param traceInfoVo
     */
    public boolean offer(final RpcTraceInfoVo traceInfoVo) {
        //如果未开启,直接返回成功,不放入队列里
        if(!eagleMonitorESConfig.isEnable()){
            return true;
        }
        boolean success =false;
        try {
            success = queue.offer(traceInfoVo);
            if (!success){
                fail.incrementAndGet();
            }
            total.incrementAndGet();
        } catch (Exception ex) {
            String format = String.format("eagle-eye-transfer offer queue exception:%s", ex.getMessage());
            log.error(format, ex);
        }
        return success;
    }

    /**
     * 提取队首的鹰眼数据
     */
    public String poll() {
        String json = null;
        RpcTraceInfoVo bean = queue.poll();
        if(bean!=null){
            try {
                json = FastJsonConvert.convertObjectToJSONWithNullValue(bean);
            } catch (Exception ex) {
                String format = String.format("eagle-eye-transfer object to json exception:%s", ex.getMessage());
                log.error(format, ex);
            }
        }
        return json;
    }

    public String clear(){
        this.queue.clear();
        return getJsonStatus();
    }

    public String status(){
        return getJsonStatus();
    }

    private String getJsonStatus() {
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("total", total.intValue());
        jsonObject.put("fail-offer", fail.intValue());
        jsonObject.put("remain", queue.size());
        jsonObject.put("ip", HostUtil.getIp());
        return jsonObject.toJSONString();
    }

    class EagleESMonitorQueueTask implements Runnable{
        @Override
        public void run() {
            boolean start = true;
            while (start) {
                try {
                    log.debug("EagleEye开始获取日志队列中数据.....");
                    int count=0;
                    List<String> jsonStringList = new ArrayList<>();
                    String jsonString=poll();
                    while(jsonString!=null){
                        jsonStringList.add(jsonString);
                        jsonString=poll();
                        //达到推送阈值大小,调用推送
                        if (jsonStringList.size() == eagleMonitorESConfig.getBatchSize()) {
                            this.pushElasticSearch(jsonStringList);
                            count+=jsonStringList.size();
                            jsonStringList.clear();
                        }
                    }
                    //把最后不满阈值大小的再次发送
                    if(!jsonStringList.isEmpty()){
                        this.pushElasticSearch(jsonStringList);
                        count+=jsonStringList.size();
                    }
                    log.debug(String.format("EagleEye结束获取日志队列中数据[%s条],睡眠%s秒钟",count, eagleMonitorESConfig.getSleepMillis()/1000));
                    Thread.sleep(eagleMonitorESConfig.getSleepMillis());
                } catch (Exception e) {
                    log.error(String.format("定时推送EagleEye monitor error : %s", e.getMessage()));
                }
            }
        }

        /**
         * <br>
         * 描 述：调用elasticSearch推送EagleEye monitor <br>
         * 作 者：hy <br>
         * 历 史: (版本) 作者 时间 注释
         *
         * @param jsonStringList
         */
        private void pushElasticSearch(List<String> jsonStringList) {
            if(jsonStringList!=null&&!jsonStringList.isEmpty()){
                SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy.MM.dd");
                String es_index_name=eagleMonitorESConfig.getIndexName()+"-"+dateFormat.format(new Date());
                // 失败最多重试3次
                for (int i = 0; i < eagleMonitorESConfig.getTryTimes(); i++) {
                    try {
                        esUtil.insertBatchDocument(es_index_name, eagleMonitorESConfig.getTypeName(), jsonStringList);
                    } catch (Exception ex) {
                        log.error(String.format("eagle-eye-transfer Error：%s", ex.getMessage()));
                        continue;
                    }

                    log.debug(String.format("eagle-eye-transfer elastic index:%s ,list size :%s", eagleMonitorESConfig.getIndexName(),
                            jsonStringList.size()));
                    break;
                }
            }
        }
    }

}
